package akka.essentials
import akka.actor.SupervisorStrategy.{Escalate, Restart, Resume, Stop}
import akka.actor._
import akka.util.Timeout
import akka.pattern.ask

import scala.collection.mutable
import scala.concurrent.Await
import scala.concurrent.duration._

/**
  * Created by shichen on 2017/6/14.
  */

case class Result()
case class DeadWorker()
case class RegisterWorker(val worker: ActorRef,val supervisor: ActorRef)

class WorkActor extends Actor with ActorLogging {
  var state: Int = 0

  override def preStart(): Unit = {
    log.info("starting workactor hashcode #{}",this.hashCode())
  }

  override def postStop(): Unit = {
    log.info("stopping workactor hashcode #{}",this.hashCode())
  }

  override def receive: Receive = {
    case value: Int =>
      if (value <= 0) throw new ArithmeticException("negative")
      else if (value == 1) context.stop(self) //这是在supervise例子中使用stop,
      // 这是正常停止，父actor并不会使之重启，如果要想让父actor使之重启，可以给自己发Kill/PoisonPill
      else state = value
    case result: Result =>
      sender() ! state
    case ex: NullPointerException =>
      throw new NullPointerException("null value")
    case _ =>
      throw new IllegalArgumentException("wrong argument")
  }
}

class SupervisorActor extends Actor with ActorLogging {
  val childActor = context.actorOf(Props[WorkActor],name="workactor")

  /* 如果没有定义它，默认处理行为是：重启*/
  override val supervisorStrategy = OneForOneStrategy(
    maxNrOfRetries = 10,withinTimeRange = 10 seconds) {
    case _: ArithmeticException => Resume
    case _: NullPointerException => Restart
    case _: IllegalArgumentException => Stop
    case _: Exception => Escalate
  }

  override def receive: Receive = {
    case result: Result =>
      childActor.tell(result,sender)
    case msg: Object =>
      childActor ! msg
  }
}

//////////////////////////////////////

class WorkerActor(ref: ActorRef) extends Actor with ActorLogging {
  var state: Int = 0

  override def preStart(): Unit = {
    log.info("staring workeractor instance hashcode # {}",this.hashCode())
    ref ! RegisterWorker(self,context.parent)
  }

  override def postStop(): Unit = {
    log.info("stopping worker instance hashcode # {}",this.hashCode())
  }

  override def receive: Receive = {
    case value: Int =>
      state = value
    case _: Result =>
      sender() ! state
    case _ =>
      context.stop(self)
      log.info("stop self")
  }
}

class MonitorActor extends Actor with ActorLogging {
  var monitorActors = new mutable.HashMap[ActorRef,ActorRef]()

  override def receive: Receive = {
    case t: Terminated =>
      if (monitorActors.contains(t.actor)) {
        log.info("recv worker actor termination msg => " + t.actor.path)
        log.info("sending msg to supervisor")
        val value: Option[ActorRef] = monitorActors.get(t.actor)
        value.get ! new DeadWorker
      }
    case msg: RegisterWorker =>
      //收到有actor注册，watch它，且记录其supervisor
      context.watch(msg.worker)
      monitorActors += (msg.worker -> msg.supervisor)
    case DeadWorker =>
      println("recv DeadWorker")
  }
}

object ch06_monitorAndSupervise extends App{
  val sys = ActorSystem("faulttolerance")
  val log = sys.log
  val value: Int = 0

  //watchTest()
  superviseTest()

  def superviseTest() {
    val supervisor = sys.actorOf(Props[SupervisorActor],
      name = "supervisor")
    log.info("sending value 8 ,no exception should be thrown")
    var msg: Int = 8
    supervisor ! msg

    implicit val timeout = Timeout(5 seconds)
    var future = (supervisor ? new Result).mapTo[Int] //?返回future[Any]
    var res = Await.result(future, timeout.duration)
    log.info("value received 1 -> {}", res)

    msg = 1
    supervisor ! msg

    log.info("sending -8, ArithmeticException thrown, supervisor decide resume")
    msg = -8
    supervisor ! msg
    future = (supervisor ? new Result).mapTo[Int]
    res = Await.result(future, timeout.duration)
    log.info("value received 2 -> {}", res)

    log.info("send null. supervisor decide restart")
    supervisor ! new NullPointerException
    future = (supervisor ? new Result).mapTo[Int]
    res = Await.result(future, timeout.duration)
    log.info("value received 3 -> {}", res)

    log.info("sending String, supervisor decide stop")
    supervisor ? "Do Something"
  }

  def watchTest(): Unit = {
    val monitor = sys.actorOf(Props[MonitorActor],"monitor")
    val worker = sys.actorOf(Props(new WorkerActor(monitor)),"worker")

    Thread.sleep(1000)

    worker ! 1

    implicit val timeout = Timeout(5 seconds)
    var future = (worker ? new Result).mapTo[Int] //?返回future[Any]
    var res = Await.result(future, timeout.duration)
    log.info("value received -> {}", res)

    worker ! "a"
  }
}
